package com.atguigu.chapter05.sink;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/17 10:44
 */
public class Flink03_Sink_Custom_Mysql {
    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
       
        env
            .socketTextStream("hadoop162", 9999)
            .map(line -> {
                String[] data = line.split(",");
                return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
            })
            .keyBy(WaterSensor::getId)
            .max("vc")
            .addSink(new MysqlSink());
            
        
        env.execute();
        
    }
    
    public static class MysqlSink extends RichSinkFunction<WaterSensor>{
    
        private PreparedStatement ps;
        private Connection conn;
    
        @Override
        public void open(Configuration parameters) throws Exception {
            // 建立到Mysql的连接
            // 1. 加载驱动
            Class.forName("com.mysql.jdbc.Driver");
            // 2. 获取连接对象
            String mysqlUrl = "jdbc:mysql://hadoop162:3306/test";
            conn = DriverManager.getConnection(mysqlUrl, "root", "aaaaaa");
            // 3. 获取预处理语句
//            String sql = "insert into sensor values(?,?,?)";
//            String sql = "insert into sensor values(?,?,?) on duplicate key update vc=?";
            String sql = "replace into sensor values(?,?,?)"; // 和前面是等价的
            ps = conn.prepareStatement(sql);
    
        }
    
        @Override
        public void close() throws Exception {
            // 关闭到Mysql的连接
            if (ps != null) {
                ps.close();
            }
    
            if (conn != null) {
                conn.close();
            }
        }
    
        // 具体的写入逻辑, 都在这个方法内实现
        @Override
        public void invoke(WaterSensor value,
                           Context context) throws Exception {
            ps.setString(1, value.getId());
            ps.setLong(2, value.getTs());
            ps.setInt(3, value.getVc());
//            ps.setInt(4, value.getVc());
    
            ps.execute();
        }
    }
}
